package com.atguigu.medical.realtime.util;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.medical.realtime.annotation.NotSink;
import com.atguigu.medical.realtime.bean.TableProcess;
import com.atguigu.medical.realtime.common.Constant;
import com.atguigu.medical.realtime.sink.PhoenixSink;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.shaded.guava30.com.google.common.base.CaseFormat;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.Properties;
import java.util.Random;
import java.util.stream.Collectors;

/**
 * @Author lzc
 * @Date 2023/4/22 09:22
 */
public class FlinkSinkUtil {
    private static final Random random = new Random();
    
    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getPhoenixSink() {
        return new PhoenixSink();
    }
    
    public static Sink<String> getKafkaSink(String topic) {
        return KafkaSink.<String>builder()
            .setBootstrapServers(Constant.KAFKA_SERVERS)
            .setRecordSerializer(
                KafkaRecordSerializationSchema.<String>builder()
                    .setTopic(topic)
                    .setValueSerializationSchema(new SimpleStringSchema())
                    .build()
            )
            .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
            // 如果要严格一次写入到 kafka 中,则必须设置事务 id 的前缀
            .setTransactionalIdPrefix(topic + "-" + random.nextLong())
            .setProperty("transaction.timeout.ms", 15 * 60 * 1000 + "")
            .build();
    }
    
    public static SinkFunction<String> getKafkaSinkFunction(String topic) {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", Constant.KAFKA_SERVERS);
        props.setProperty("transaction.timeout.ms", 15 * 60 * 1000 + "");
        return new FlinkKafkaProducer<String>(
            "default",
            new KafkaSerializationSchema<String>() {
                @Override
                public ProducerRecord<byte[], byte[]> serialize(String element,
                                                                @Nullable Long timestamp) {
                    return new ProducerRecord<>(topic, element.getBytes(StandardCharsets.UTF_8));
                }
            },
            props,
            FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }
    
    public static Sink<Tuple2<JSONObject, TableProcess>> getKafkaSink() {
        return KafkaSink.<Tuple2<JSONObject, TableProcess>>builder()
            .setBootstrapServers(Constant.KAFKA_SERVERS)
            .setRecordSerializer(
                new KafkaRecordSerializationSchema<Tuple2<JSONObject, TableProcess>>() {
                    @Override
                    public ProducerRecord<byte[], byte[]> serialize(Tuple2<JSONObject, TableProcess> t,
                                                                    KafkaSinkContext context,
                                                                    Long timestamp) {
                        
                        return new ProducerRecord<>(t.f1.getSinkTable(),
                                                    t.f0.toJSONString().getBytes(StandardCharsets.UTF_8)
                        );
                    }
                }
            )
            .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
            .setTransactionalIdPrefix("---" + random.nextLong())
            .setProperty("transaction.timeout.ms", 15 * 60 * 1000 + "")
            .build();
    }
    
    
    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getKafkaSinkOld() {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", Constant.KAFKA_SERVERS);
        props.setProperty("transaction.timeout.ms", 15 * 60 * 1000 + "");
        
        return new FlinkKafkaProducer<Tuple2<JSONObject, TableProcess>>(
            "default",
            new KafkaSerializationSchema<Tuple2<JSONObject, TableProcess>>() {
                @Override
                public ProducerRecord<byte[], byte[]> serialize(Tuple2<JSONObject, TableProcess> t,
                                                                @Nullable Long timestamp) {
                    String topic = t.f1.getSinkTable();
                    byte[] v = t.f0.toJSONString().getBytes(StandardCharsets.UTF_8);
                    return new ProducerRecord<>(topic, v);
                }
            },
            props,
            FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }
    
    public static <T> SinkFunction<T> getClickHouseSink(String table, Class<T> tClass) {
        /*
         在 jdbc 的基础, 封装一个 clickhouse sink
         */
        String driver = Constant.CLICKHOUSE_DRIVER;
        String url = Constant.CLICKHOUSE_URL;
        // insert into user(a,b,c,d)values(?,?,?,?)
        String cs = Arrays.stream(tClass.getDeclaredFields())
            .filter(field -> {
                NotSink notSink = field.getAnnotation(NotSink.class);
                return notSink == null;
            })
            .map(f -> CaseFormat.LOWER_CAMEL.to(CaseFormat.LOWER_UNDERSCORE, f.getName()))
            .collect(Collectors.joining(","));
        StringBuilder sql = new StringBuilder();
        sql
            .append("insert into ")
            .append(table)
            .append("(")
            .append(cs)  // TODO
            .append(")values(")
            .append(cs.replaceAll("[^,]+", "?"))
            .append(")");
        
        
        System.out.println("clickhouse 的插入语句: " + sql);
        return JdbcSink.sink(
            sql.toString(),
            new JdbcStatementBuilder<T>() {
                @Override
                public void accept(PreparedStatement ps,
                                   T t) throws SQLException {
                    //  insert into user(a,b,c,d)values(?,?,?,?)
                    // 给 sql 中的占位符进行赋值 TODO
                    // 得到一个 Class:
                    // 1. Class.forName("类名")  2. 类名.class  3.对象.getClass
                    Field[] fields = t.getClass().getDeclaredFields();
                    try {
                        for (int i = 0, position = 1; i < fields.length; i++) {
                            Field field = fields[i];
                            if (field.getAnnotation(NotSink.class) == null) {
                                field.setAccessible(true);
                                // 给占位符的索引使用单独的计数器.因为 i 会有跳过的现象, 占位符的索引必须是连续
                                ps.setObject(position++, field.get(t));  // 未来会有一个 bug
                                
                            }
                            
                        }
                    } catch (IllegalAccessException e) {
                        throw new RuntimeException(e);
                    }
                    
                }
            },
            new JdbcExecutionOptions.Builder()
                .withBatchSize(1000)
                .withBatchIntervalMs(3000)
                .withMaxRetries(3)
                .build(),
            new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withUrl(url)
                .withDriverName(driver)
                .withUsername("default")
                .withPassword("aaaaaa")
                .build()
        );
    }
    
}
